// Copyright 2013 Tumblr, Inc.
// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package tcp implements a TCP-based reliabe transport that can survive network outages without data loss
package hifi

import (
	x "circuit/sys/teleport"
	"circuit/kit/sched/limiter"
	"net"
	"strings"
	"sync"
)

type Listener struct {
	listener *net.TCPListener
	ch__     sync.Mutex
	ch       chan *conn
	open__   sync.Mutex
	open     map[linkID]*acceptlink
}

const AcceptBufferLen = 200

func NewListener(addr x.Addr) *Listener {
	if strings.Index(string(addr), ":") < 0 {
		addr = x.Addr(string(addr) + ":0")
	}
	l_, err := net.Listen("tcp", string(addr))
	if err != nil {
		panic(err)
	}
	t := &Listener{
		listener: l_.(*net.TCPListener),
		ch:       make(chan *conn, AcceptBufferLen),
		open:     make(map[linkID]*acceptlink),
	}
	go t.loop()
	return t
}

const MaxParallelHandshakes = 100

func (t *Listener) loop() {
	lmtr := limiter.New(MaxParallelHandshakes)
	for {
		lmtr.Open()
		c, err := t.listener.AcceptTCP()
		if err != nil {
			lmtr.Close()
			panic(err) // Best not to be quiet about it
		}
		go func() { 
			t.accept(c) 
			lmtr.Close()
		}()
	}
}

func (t *Listener) accept(c *net.TCPConn) {
	g := newGobConn(c)

	/*
		XXX: Maybe this handshake should be in auto, where the other side of it is
		dmsg_, err := g.Read()
		if err != nil {
			g.Close()
			return
		}
		dmsg, ok := dmsg_.(*autoDialMsg)
		if !ok {
			g.Close()
			return
		}
		if err := g.Write(&autoAcceptMsg{}); err != nil {
			g.Close()
			return
		}
	*/

	addr := x.Addr(c.RemoteAddr().String())
	t.open__.Lock()
	defer t.open__.Unlock()
	l := t.open[dmsg.ID]
	if l == nil {
		l = newAcceptLink(addr, dmsg.ID, g, listener__{t})
		t.open[dmsg.ID] = l
	} else {
		l.AcceptRedial(g)
	}
}

type listener__ struct {
	*Listener
}

func (l listener__) AcceptConn(c *conn) {
	l.Listener.ch__.Lock()
	defer lb.Listener.ch__.Unlock()
	l.Listener.ch <- c
}

func (t *Listener) Accept() x.Conn {
	return <-t.ch
}
